package com.bitmain.mmst.miner.monitor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.BooleanUtils;

import com.bitmain.mmst.domain.MinerNetworkMessage;
import com.bitmain.mmst.domain.MinerShelf;
import com.bitmain.mmst.domain.MinerState;
import com.bitmain.mmst.feature.time.TimeService;
import com.bitmain.mmst.miner.TaskExecuteEvent;
import com.bitmain.mmst.miner.monitor.executor.MinerNetworkExecutor;
import com.bitmain.mmst.miner.monitor.helper.DetectDataHelper;
import com.bitmain.mmst.miner.monitor.internal.MonitorOption;
import com.bitmain.ulnar.aider.DateAider;

/**
 * 
 *
 * @author Richard - rui.wu@bitmain.com
 * @since 2016年7月5日
 *
 */
public class MinerNetworkDetector extends MinerMonitor{
	/**
	 * 所有机架IP范围内的总数
	 */
	private AtomicInteger numAllLimit;
	/**
	 * 所有机架忽略的IP总数
	 */
	private AtomicInteger numAllIgnore;
	/**
	 * 所有机架排除忽略的IP的总数
	 */
	private AtomicInteger numAllTotal;
	/**
	 * 所有机架能ping通的矿机总数
	 */
	private AtomicInteger numAllPinged;
	/**
	 * 所有机架能获取到MAC地址的矿机总数
	 */
	private AtomicInteger numAllMaced;
	/**
	 * 所有机架丢失（ping不通）的矿机总数
	 */
	private AtomicInteger numAllMissed;
	/**
	 * 检测时间
	 */
	private long detectTime;
	
	public MinerNetworkDetector() {
		this.numAllLimit = new AtomicInteger(0);
		this.numAllIgnore = new AtomicInteger(0);
		this.numAllTotal = new AtomicInteger(0);
		this.numAllPinged = new AtomicInteger(0);
		this.numAllMaced = new AtomicInteger(0);
		this.numAllMissed = new AtomicInteger(0);
	}
	void execute(){
		if(numNetworkDetector()>0)
			return;
		this.detectTime = TimeService.getTimeMillis();
		incrementNetworkDetector();
		
		System.out.println("【网络状态检测】1、NetworkDetector 矿机网络状态检测" + DateAider.formatMillis(detectTime) + " - 开始ʼ");
		
		List<MinerShelf> shelfs = DetectDataHelper.fetchAllShelfs(); 
		
		if(CollectionUtils.isEmpty(shelfs)){
			System.out.println("【网络状态检测】2、没有获取到机架");
			decrementNetworkDetector();
			return;
		}else{
			System.out.println("【网络状态检测】2、获取到机架，机架长度为："+shelfs.size());
		}
			
		List<MinerNetworkMessage> messages = new ArrayList<MinerNetworkMessage>();
		
		for(MinerShelf shelf :shelfs){
			String ipBlock = shelf.getIpBlock();
			int ipBeginning = shelf.getIpBeginning();
			int ipEnd = shelf.getIpEnd();
			if(ipBeginning > ipEnd){
				int temp = ipBeginning;
				ipBeginning = ipEnd;
				ipEnd = temp;
			}
			Integer[] ipIgnore = shelf.getIgnores();
			int numLimit = ipEnd - ipBeginning + 1;
			int numIgnore = 0;
			if(ArrayUtils.isNotEmpty(ipIgnore)){
				numIgnore = ipIgnore.length;
			}
			int numTotal = numLimit - numIgnore;
			
			
			numAllLimit.addAndGet(numLimit);
			numAllIgnore.addAndGet(numIgnore);
			numAllTotal.addAndGet(numTotal);
			for(int i=ipBeginning;i<=ipEnd;i++){
				if(numIgnore > 0&&ArrayUtils.contains(ipIgnore, i)){
					continue;
				}
				MinerNetworkMessage message = new MinerNetworkMessage();
				message.setShelfId(shelf.getId());
				message.setShelfName(shelf.getName());
				message.setHost(ipBlock+"."+i);
				message.setIpIndex(i);
				messages.add(message);
			}
		}
		allocateExecutor(messages);
	}
	private void allocateExecutor(List<MinerNetworkMessage> messages) {
		if(CollectionUtils.isEmpty(messages)){
			return;
		}
		int numThread = MonitorOption.networkThreadSize;
		if(messages.size() <= numThread){
			executeOverall(messages , numThread);
		}else{
			executeBatching(messages , numThread);
		}
	}
	private void executeOverall(List<MinerNetworkMessage> messages, int numThread) {
		System.out.println("【网络状态检测】3、矿机总数为："+messages.size()+"，最大线程数为："+numThread+"，执行executeOverall方法");
		ExecutorService service = Executors.newFixedThreadPool(numThread);
		for(MinerNetworkMessage message:messages){
			submitExecutor(message , service);
		}
		service.shutdown();
		long start = System.currentTimeMillis();
		boolean done = false;
		while(!done){
			if(service.isTerminated()){
				service.shutdownNow();
				done = true;
				//System.out.println("【完成】检测");
			}else{
				if(System.currentTimeMillis() - start > 60*1000){
					service.shutdownNow();
					done = true;
					System.out.println("【网络状态检测】4、检测超时");
				}
			}
			try {
				Thread.sleep(1000);
			} catch (Exception e) {
			}
		}
		executeDone();
	}
	
	private void executeBatching(List<MinerNetworkMessage> messages, int numThread) {
		int numBatch = (messages.size() + numThread -1) / numThread;
		
		System.out.println("【网络状态检测】3、矿机总数为："+ messages.size()+"，最大线程数为："+numThread+"，执行executeBatching方法,分批数为："+numBatch);
		for(int i=0;i<numBatch;i++){
			int fromIndex = i*numThread;//0,100,200
			int toIndex = fromIndex + numThread ;//99,100
			if(toIndex >= messages.size()){
				toIndex = messages.size();
			}
			System.out.println("【网络状态检测】检测批次    第" + (i+1) +"批，从"+fromIndex+"--"+toIndex);
			List<MinerNetworkMessage> batchMessages = messages.subList(fromIndex, toIndex);
			ExecutorService service = Executors.newFixedThreadPool(numThread);
			for(MinerNetworkMessage message:batchMessages){
				submitExecutor(message , service);
			}
			service.shutdown();
			long start = System.currentTimeMillis();
			boolean done = false;
			while(!done){
				if(service.isTerminated()){
					service.shutdownNow();
					done = true;
				}else{
					if(System.currentTimeMillis() - start > 1000*60){
						service.shutdownNow();
						done = true;
						System.out.println("【网络状态检测】4、检测超时");
					}
				}
				try {
					Thread.sleep(500);
				} catch (Exception e) {
				}
			}
		}
		executeDone();
	}
	private void submitExecutor(MinerNetworkMessage message, ExecutorService service) {
		MinerNetworkExecutor executor = new MinerNetworkExecutor(message);
		CompletableFuture<MinerNetworkMessage> future = CompletableFuture.supplyAsync(executor, service);
		future.thenAccept(m -> processMessage(m));
	}
	
	private void processMessage(MinerNetworkMessage message) {
		MinerState minerState = MinerStateCache.INSTANCE.pull(message.getHost());
		if (minerState == null) {
			MinerState state = new MinerState();
			state.setShelfId(message.getShelfId());
			state.setShelfName(message.getShelfName());
			state.setHost(message.getHost());
			state.setIpIndex(message.getIpIndex());
			state.setMac(message.getMac());
			state.setPingable(message.getPingable());
			state.setMacable(message.getMacable());
			state.setDetectTime(detectTime);
			MinerStateCache.INSTANCE.put(state);
		} else {
			minerState.setShelfId(message.getShelfId());
			minerState.setShelfName(message.getShelfName());
			minerState.setIpIndex(message.getIpIndex());
			minerState.setMac(message.getMac());
			minerState.setPingable(message.getPingable());
			minerState.setMacable(message.getMacable());
			minerState.setDetectTime(detectTime);
		}
		
		if(BooleanUtils.isTrue(minerState.getMacable())){
			numAllMaced.incrementAndGet();
		}
		if(BooleanUtils.isTrue(minerState.getPingable())){
			numAllPinged.incrementAndGet();
		}else{
			numAllMissed.incrementAndGet();
		}
	}
	
	private void executeDone() {
		System.out.println("【网络状态检测】4、执行executeDone方法");
		
		long elapsedTime = TimeService.getTimeMillis() - detectTime;
		//打印检测信息
		System.out.println();
		System.out.println("【网络状态检测】5、汇总信息：-----------------");
		System.out.println("检测初始时间:"+DateAider.formatMillis(detectTime));
		System.out.println("检测耗时:"+DateAider.formatMillis(elapsedTime,"mm:ss"));
		System.out.println("检测ip数:"+numAllLimit.get());
		System.out.println("检测忽略ip数:"+numAllIgnore.get());
		System.out.println("检测实际ip数:"+numAllTotal.get());
		System.out.println("获取到mac地址数:"+numAllMaced.get());
		System.out.println("ping通数:"+numAllPinged.get());
		System.out.println("检测丢失数:"+numAllMissed.get());
		System.out.println();
		System.out.println("【网络状态检测】6、网络状态检测执行完成时，缓存个数为："+MinerStateCache.INSTANCE.getPoisedMinerCache().size());
		
		decrementNetworkDetector();
		
		//执行完成，出发监听器
		TaskExecuteEvent event = new TaskExecuteEvent();
		executeDone(event);
	}
}
